package yuekao8.ads;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import yuekao8.entity.Tm5_1;
import yuekao8.entity.Tm5_3;
import yuekao8.util.KafkaUtil;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.HashSet;

public class WindowsTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //从 Kafka 订单明细主题读取数据，分组开窗聚合，统计各维度（省份、商品）各窗口的订单数、订单金额、商品数量，补全维度信息，将数据写入 ClickHouse 交易域SKU粒度下单各窗口汇总表。
        //5.1) 编写Flink流式程序，从Kafka对象实时消费DWS层交易域下单事务事实表数据，设置允许乱序最大水位线为1秒，提取相关业务字段值，封装Java实体类对象；（5分）
//        env.addSource(KafkaUtil.kafkaSources("dws_trade_orders")).print();
        SingleOutputStreamOperator<Tm5_1> tm5_1data = env.addSource(KafkaUtil.kafkaSource("dws_trade_orders")).map(new MapFunction<String, Tm5_1>() {
            @Override
            public Tm5_1 map(String s) throws Exception {
                //Tuple6<OrderInfo, OrderDetail, ProductSpu, ProductSku, Region, Shop>
                //int id;//各窗口的订单数
                Integer id = JSON.parseObject(s).getJSONObject("f0").getInteger("id");
                //int region_id;//省份id
                Integer region_id = JSON.parseObject(s).getJSONObject("f5").getInteger("region_id");
                //String region_name;//省份
                String region_name = JSON.parseObject(s).getJSONObject("f4").getString("name");
                //int product_spu_id;//商品id
                Integer product_spu_id = JSON.parseObject(s).getJSONObject("f3").getInteger("product_spu_id");
                //String product_spu_name;//商品
                String product_spu_name = JSON.parseObject(s).getJSONObject("f2").getString("name");
                //Double amount;//订单金额
                Double amount = JSON.parseObject(s).getJSONObject("f1").getDouble("amount");
                //int sku_num;//商品数量
                Integer sku_num = JSON.parseObject(s).getJSONObject("f1").getInteger("sku_num");
                //String create_time;
                String create_time = JSON.parseObject(s).getJSONObject("f0").getString("create_time");
                return new Tm5_1(id, region_id, region_name, product_spu_id, product_spu_name, amount, sku_num, create_time);
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy
                .<Tm5_1>forBoundedOutOfOrderness(Duration.ofSeconds(1))
                .withTimestampAssigner((event, timestamp) -> {
                    try {
                        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(event.getCreate_time()).getTime();
                    } catch (ParseException e) {
                        throw new RuntimeException(e);
                    }
                }));
//        tm5_1data.print();
        //5.2)设置事件时间窗口为1天，每隔1秒触发窗口计算，实时累加统计各个维度的指标度量中（订单数据、订单金额、商品数据量）；（5分）
        SingleOutputStreamOperator<Tm5_3> process = tm5_1data.keyBy(new KeySelector<Tm5_1, Tuple4<Integer, String, Integer, String>>() {
                    @Override
                    public Tuple4<Integer, String, Integer, String> getKey(Tm5_1 o) throws Exception {
                        return new Tuple4<>(o.getRegion_id(), o.getRegion_name(), o.getProduct_spu_id(), o.getProduct_spu_name());
                    }
                }).window(SlidingEventTimeWindows.of(Time.days(1), Time.seconds(1)))
                .process(new ProcessWindowFunction<Tm5_1, Tm5_3, Tuple4<Integer, String, Integer, String>, TimeWindow>() {
                    @Override
                    public void process(Tuple4<Integer, String, Integer, String> one,
                                        ProcessWindowFunction<Tm5_1, Tm5_3, Tuple4<Integer, String, Integer, String>, TimeWindow>.Context context,
                                        Iterable<Tm5_1> iterable,
                                        Collector<Tm5_3> collector) throws Exception {
                        //订单数据、订单金额、商品数据量
                        HashSet<Integer> set = new HashSet<>();
                        Double sumamount = 0.0;
                        Integer sum = 0;
                        for (Tm5_1 o : iterable) {
                            set.add(o.getId());
                            sumamount += o.getAmount();
                            sum += o.getSku_num();
                        }
                        String sta = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(context.window().getStart());
                        String end = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(context.window().getEnd());
                        collector.collect(new Tm5_3(sta, end, set.size(), sumamount, sum, one.f1, one.f3));
                    }
                });
        process.print();
        //5.3)将上述交易域SKU粒度下单各窗口汇总结果数据，实时存储Clickhouse表，其中表的引擎ReplacingMergeTree主键相同时，更新字段值；（5分）
        //备注：答题截图时，包括ClickHouse中创建表语句，查看表的数据等完整信息，否则0分处理。
        env.execute();
    }
}
